package com.tdcy.framework.util;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import com.tdcy.framework.exception.BaseException;

/**
 * 任务控制器.
 */
@SuppressWarnings(value={"unchecked", "rawtypes"}) 
public class TaskController {

	/** The Constant threadPoolLock. */
	private static final Object threadPoolLock = new Object();

	/** The core pool size. */
	private static int corePoolSize = 1;

	/** The max pool size. */
	private static int maxPoolSize = 5;

	/** The keep alive seconds. */
	private static int keepAliveSeconds = 300;// 5分钟

	/** The allow core thread time out. */
	private static boolean allowCoreThreadTimeOut = false;

	/** The queue capacity. */
	private static int queueCapacity = 10000;// Integer.MAX_VALUE;

	/** The rejected execution handler. */
	private static RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.CallerRunsPolicy();

	/** The wait for tasks to complete on shutdown. */
	private static boolean waitForTasksToCompleteOnShutdown = true;

	/** The thread pool executor. */
	private static ThreadPoolExecutor threadPoolExecutor;

	/**
	 * 设置核心线程数.
	 * 
	 * @param corePoolSize
	 *            the new core pool size
	 */
	public static void setCorePoolSize(final int corePoolSize) {
		TaskController.corePoolSize = corePoolSize;
	}

	/**
	 * 设置最大线程数.
	 * 
	 * @param maxPoolSize
	 *            the new max pool size
	 */
	public static void setMaxPoolSize(final int maxPoolSize) {
		TaskController.maxPoolSize = maxPoolSize;
	}

	/**
	 * 设置线程保持活动的时间.
	 * 
	 * @param keepAliveSeconds
	 *            the new keep alive seconds
	 */
	public static void setKeepAliveSeconds(final int keepAliveSeconds) {
		TaskController.keepAliveSeconds = keepAliveSeconds;
	}

	/**
	 * 获取核心线程数.
	 * 
	 * @return the core pool size
	 */
	public static int getCorePoolSize() {
		return corePoolSize;
	}

	/**
	 * 获取最大线程数.
	 * 
	 * @return the max pool size
	 */
	public static int getMaxPoolSize() {
		return maxPoolSize;
	}

	/**
	 * 获取线程保持活动的时间.
	 * 
	 * @return the keep alive seconds
	 */
	public static int getKeepAliveSeconds() {
		return keepAliveSeconds;
	}

	/**
	 * 获取是否允许核心线程超时.
	 * 
	 * @return the allow core thread time out
	 */
	public static boolean getAllowCoreThreadTimeOut() {
		return allowCoreThreadTimeOut;
	}

	/**
	 * 获取是否等待任务结束.
	 * 
	 * @param waitForJobsToCompleteOnShutdown
	 *            the wait for jobs to complete on shutdown
	 * @return the wait for tasks to complete on shutdown
	 */
	public static boolean getWaitForTasksToCompleteOnShutdown(final boolean waitForJobsToCompleteOnShutdown) {
		return waitForTasksToCompleteOnShutdown;
	}

	/**
	 * 初始化.
	 */
	protected static void init() {
		if (threadPoolExecutor != null) {
			return;
		}

		BlockingQueue queue = createQueue(queueCapacity);
		threadPoolExecutor = new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveSeconds, TimeUnit.SECONDS, queue, rejectedExecutionHandler);
		if (allowCoreThreadTimeOut) {
			threadPoolExecutor.allowCoreThreadTimeOut(true);
		}
	}

	/**
	 * 创建队列.
	 * 
	 * @param queueCapacity
	 *            the queue capacity
	 * @return the blocking queue
	 */
	protected static BlockingQueue createQueue(final int queueCapacity) {
		if (queueCapacity > 0) {
			return new LinkedBlockingQueue(queueCapacity);
		} else {
			return new SynchronousQueue();
		}
	}

	/**
	 * 获取线程池执行器.
	 * 
	 * @return the thread pool executor
	 * @throws IllegalStateException
	 *             the illegal state exception
	 */
	public static ThreadPoolExecutor getThreadPoolExecutor() throws IllegalStateException {
		if (threadPoolExecutor == null) {
			synchronized (threadPoolLock) {
				init();
			}
		}

		return threadPoolExecutor;
	}

	/**
	 * 执行任务.
	 * 
	 * @param task
	 *            the task
	 */
	public static void execute(final Runnable task) {
		Executor executor = getThreadPoolExecutor();
		try {
			executor.execute(task);
		} catch (RejectedExecutionException ex) {
			throw new BaseException("任务执行器 [" + executor + "] 不能接收任务: " + task, ex);
		}
	}

	/**
	 * 移除任务.
	 * 
	 * @param task
	 *            the task
	 * @return true, if successful
	 */
	public static boolean remove(final Runnable task) {
		return getThreadPoolExecutor().remove(task);
	}

	/**
	 * 返回线程池大小.
	 * 
	 * @return the pool size
	 * @see java.util.concurrent.ThreadPoolExecutor#getPoolSize()
	 */
	public static int getPoolSize() {
		return getThreadPoolExecutor().getPoolSize();
	}

	/**
	 * 返回活动线程数.
	 * 
	 * @return the active count
	 * @see java.util.concurrent.ThreadPoolExecutor#getActiveCount()
	 */
	public static int getActiveCount() {
		return getThreadPoolExecutor().getActiveCount();
	}

	/**
	 * 获取已完成任务数.
	 * 
	 * @return the completed task count
	 */
	public static long getCompletedTaskCount() {
		return getThreadPoolExecutor().getCompletedTaskCount();
	}

	/**
	 * 停止服务(框架调用）.
	 */
	public static void shutdown() {
		synchronized (threadPoolLock) {
			if (threadPoolExecutor == null) {
				return;
			}

			if (waitForTasksToCompleteOnShutdown) {
				threadPoolExecutor.shutdown();
			} else {
				threadPoolExecutor.shutdownNow();
			}
		}
	}
}